/*
 * 版权所有 (C) 2015 知启蒙(ZHIQIM) 保留所有权利。[遇见知启蒙，邂逅框架梦]
 * 
 * https://zhiqim.org/project/zhiqim_framework/zhiqim_kernel.htm
 *
 * Zhiqim Kernel is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */
package org.zhiqim.kernel.websocket;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import javax.net.ssl.KeyManager;
import javax.net.ssl.TrustManager;

import org.zhiqim.kernel.constants.HttpConstants;
import org.zhiqim.kernel.control.Threadx;
import org.zhiqim.kernel.logging.Log;
import org.zhiqim.kernel.logging.LogFactory;
import org.zhiqim.kernel.model.HttpLine;
import org.zhiqim.kernel.model.codes.SSL;
import org.zhiqim.kernel.model.codes.URI;
import org.zhiqim.kernel.model.maps.HashMapSS;
import org.zhiqim.kernel.schedule.Interval;
import org.zhiqim.kernel.schedule.Task;
import org.zhiqim.kernel.util.Closes;
import org.zhiqim.kernel.util.Ints;
import org.zhiqim.kernel.util.Sockets;
import org.zhiqim.kernel.util.Strings;

/**
 * WebSocket客户端类，使用方式如：
 * 
 * WebSocket conn = new WebSocket("http://127.0.0.1/service/ws");
 * conn.setProtocol("zoa_announcement");
 * conn.setService(new WebSocketServiceImpl());
 * conn.execute();
 * if (!conn.isResponseSuccess())
 * {
 *     System.out.println("连接失败："+conn.getResponseText());
 *     return;
 * }
 * 
 * 注意在WebSocketServiceImpl中所有的实现方法必须[异步处理]或[能轻量快速返回]，否则会阻塞业务
 *
 * @version v1.5.2 @author zouzhigang 2020-4-24 新建与整理
 */
public class WebSocket implements HttpConstants, WebSocketConstants
{
    private static final Log log = LogFactory.getLog(WebSocket.class);
    
    private final String url;                               //服务端URL
    private final URI uri;                                  //服务端URL解析成URI
    private final AtomicInteger handshakeNum;               //当前握手次数
    
    private int handshakeTime;                              //设置的握手时间间隔，单位：秒
    private int handshakeMaxNum;                            //设置的握手最大次数，达到则重连
    private int maskingKey;                                 //设置的消息掩码
    
    private HttpLine liner;                                 //请求头处理器
    private HashMapSS requestPropertyMap;                   //请求属性参数表

    private String sslProtocol = "TLSv1.2";                 //SSL协议
    private KeyManager[] sslKeyManagers;                    //SSL密钥管理器
    private TrustManager[] sslTrustManagers;                //SSL信任管理器
    
    private WebSocketThread thread;                         //连接成功监听线程
    private Interval interval;                              //连接检查任务
    private WebSocketService service;                       //连接成功后的业务侧回调服务
    
    //客户端连接和输入输出流
    private Socket socket;                                  //当前SOCKET
    private InputStream input;                              //当前输入流
    private OutputStream output;                            //当前输出流
    
    private int responseStatus;                             //响应状态码
    private String responseText;                            //响应状态信息

    private String responseStatusLine;                      //响应状态行
    private HashMapSS responsePropertyMap;                  //响应属性参数表
    
    /** 标准构造函数，提供URL */
    public WebSocket(String url)
    {
        this.url = url;
        this.uri = new URI();
        this.handshakeNum = new AtomicInteger();
        
        this.handshakeTime = 5;
        this.handshakeMaxNum = 3;
        this.maskingKey = 13;
        this.service = new WebSocketServicer();
        this.liner = new HttpLine();
        this.thread = new WebSocketThread(this);
        
        this.requestPropertyMap = new HashMapSS();
        this.addDefaultRequestProperty();
        
        this.responsePropertyMap = new HashMapSS();
    }

    /** 增加缺省请求属性 */
    private void addDefaultRequestProperty()
    {
        requestPropertyMap.put(_CONNECTION_, _UPGRADE_);
        requestPropertyMap.put(_UPGRADE_, _WEBSOCKET_);
        requestPropertyMap.put(_SEC_WEBSOCKET_KEY_, "dGhlIHNhbXBsZSBub25jZQ==");
        requestPropertyMap.put(_SEC_WEBSOCKET_VERSION_, "13");
    }
    
    /************************************************************************/
    //属性
    /************************************************************************/
    
    /** 设置SSL协议 */
    public WebSocket setSSLProtocol(String protocol)
    {
        this.sslProtocol = protocol;
        return this;
    }
    
    /** 设置SSL密钥管理器 */
    public WebSocket setKeyManager(KeyManager[] keyManagers)
    {
        this.sslKeyManagers = keyManagers;
        return this;
    }
    
    /** 设置SSL信任管理器 */
    public WebSocket setTrustManager(TrustManager[] trustManagers)
    {
        this.sslTrustManagers = trustManagers;
        return this;
    }
    
    /** 设置服务监听器 */
    public WebSocket setService(WebSocketService service)
    {
        this.service = service;
        return this;
    }
    
    /** 设置掩码值 */
    public WebSocket setMaskingKey(int maskingKey)
    {
        this.maskingKey = maskingKey;
        return this;
    }
    
    /** 设置握手参数 */
    public WebSocket setHandshake(int handshakeSecond, int handshakeMaxNum)
    {
        this.handshakeTime = handshakeSecond;
        this.handshakeMaxNum = handshakeMaxNum;
        return this;
    }
    
    /** 设置协议 */
    public WebSocket setProtocol(String protocol)
    {
        requestPropertyMap.put(_SEC_WEBSOCKET_PROTOCOL_, protocol);
        return this;
    }
    
    /** 增加请求属性 */
    public WebSocket addRequestProperty(String key, int value)
    {
        addRequestProperty(key, String.valueOf(value));
        return this;
    }
    
    /** 增加请求属性 */
    public WebSocket addRequestProperty(String key, String value)
    {
        requestPropertyMap.put(key, value);
        return this;
    }
    
    /** 判断是否已连接（要求线程已开且响应正确） */
    public final boolean isConnected()
    {
        return thread.isRunning() && isResponseSuccess();
    }
    
    /************************************************************************/
    //执行&关闭
    /************************************************************************/

    /** 执行WebSocket */
    public final synchronized void execute()
    {
        if (thread.isRunning())
        {//已运行的直接返回
            return;
        }
        
        if (responseStatus == _70_MALFORMED_URL_)
        {//请求URL不正确
            return;
        }
        
        if (responseStatus == 0 && !uri.parse(url).isParsed())
        {//第一次解析URL
            responseStatus = _70_MALFORMED_URL_;
            responseText = "请求的URL不正确";
            return;
        }
        
        if (interval == null)
        {//第一次启动定时任务
            interval = Interval.shedule(new WebSocketTask(this), handshakeTime * 1000);
        }
        
        String host = uri.getHost();
        int port = uri.getPort();
        
        if (!requestPropertyMap.containsKey(_HOST_))
        {//没有指定HOST的，增加URL的HOST到请求属性中
            requestPropertyMap.put(_HOST_, host);
        }
        
        try
        {
            InetSocketAddress endpoint = new InetSocketAddress(host, port);
            
            Socket socket = null;
            if (uri.isHttps())
            {//HTTPS
                socket = SSL.getSocketFactory(sslProtocol, sslKeyManagers, sslTrustManagers).createSocket();
            }
            else
            {//HTTP
                socket = new Socket();
            }
            
            Sockets.setOption(socket, true, 0, 0, true);
            socket.connect(endpoint, 10 * 1000);
            
            input = socket.getInputStream();
            output = socket.getOutputStream();
            
            //写入HTTP升级协议请求
            doWriteRequestContent();
            
            //读出HTTP升级协议响应
            doReadResponseProperty();
            
            if (!isResponseSuccess())
            {//响应失败
                log.error("WebSocket连接[%s:%s]失败：[%s]", host, port, responseStatusLine);
            }
            else
            {//响应正常则启动监听线程并通知服务onopen
                log.info("WebSocket连接[%s:%s]成功", host, port);
                thread.open();
                service.onOpen(this);
            }
        }
        catch (SocketTimeoutException e) 
        {
            responseStatus = _73_SOCKET_TIMEOUT_;
            responseText = Strings.format("WebSocket连接[%s:%s]超时", host, port);
            log.error(responseText);
        }
        catch (ConnectException e)
        {
            responseStatus = _91_CONNECT_EXCEPTION_;
            responseText = Strings.format("WebSocket连接[%s:%s]ConnectException异常：[%s]", host, port, e.getMessage());
            log.error(responseText);
        }
        catch (SocketException e)
        {
            responseStatus = _90_SOCKET_EXCEPTION_;
            responseText = Strings.format("WebSocket连接[%s:%s]SocketException异常：[%s]", host, port, e.getMessage());
            log.error(responseText);
        }
        catch (IOException e)
        {
            responseStatus = _98_IO_EXCEPTION_;
            responseText = Strings.format("WebSocket连接[%s:%s]IOException异常：[%s]", host, port, e.getMessage());
            log.error(responseText);
        }
        catch(Exception e)
        {
            responseStatus = _99_EXCEPTION_;
            responseText = Strings.format("WebSocket连接[%s:%s]%s异常：[%s]", host, port, e.getClass().getName(), e.getMessage());
            log.error(responseText);
        }
    }
    
    /** 写入请求内容 */
    private void doWriteRequestContent() throws IOException
    {
        StringBuilder strb = new StringBuilder();
        //PATH
        strb.append(_GET_).append(_ONE_).append(uri.getPathQuery()).append(_ONE_).append(_HTTP_1_1_).append(_BR_);
        //必须的属性
        for (String key : requestPropertyMap.keySet())
        {//其他的用户属性
            strb.append(key).append(":").append(requestPropertyMap.get(key)).append(_BR_);
        }
        //结束符
        strb.append(_BR_);
        
        output.write(Strings.toUTF8(strb.toString()));
    }
    
    /** 读出响应属性  */
    private void doReadResponseProperty() throws IOException
    {
        int status = liner.parse(input);
        if (status != 0)
        {
            responseStatus = status;
            responseText = "服务端请求有语法错误，不能被客户端所理解["+status+"]";
            return;
        }
        
        List<String> list = liner.list();
        if (list.isEmpty())
        {
            responseStatus = _82_SERVER_DATA_EXCEPTION_;
            responseText = "服务端请求有语法错误，不能被客户端所理解[list.isEmpty()]";
            return;
        }
        
        responseStatusLine = list.get(0);
        int ind = responseStatusLine.indexOf(_ONE_);
        if (ind == -1)
        {
            responseStatus = _82_SERVER_DATA_EXCEPTION_;
            responseText = "服务端请求有语法错误，不能被客户端所理解["+responseStatusLine+"]";
            return;
        }
        int ind2 = responseStatusLine.indexOf(_ONE_, ind+1);
        if (ind2 == -1)
        {
            responseStatus = _82_SERVER_DATA_EXCEPTION_;
            responseText = "服务端请求有语法错误，不能被客户端所理解["+responseStatusLine+"]";
            return;
        }
        responseStatus = Ints.toInt(responseStatusLine.substring(ind+1, ind2), _82_SERVER_DATA_EXCEPTION_);
        if (responseStatus == _82_SERVER_DATA_EXCEPTION_)
        {
            responseText = "服务端请求有语法错误，不能被客户端所理解["+responseStatusLine+"]";
            return;
        }
        
        for (int i=1;i<list.size();i++)
        {
            String line = list.get(i);
            int idx = line.indexOf(":");
            responsePropertyMap.put(line.substring(0, idx).trim().toLowerCase(), line.substring(idx+1).trim());
        }
    }
    
    /** 关闭WebSocket */
    public final synchronized void close()
    {
        if (interval != null)
        {//关闭定时任务
            interval.close();
            interval = null;
        }
        
        if (thread != null)
        {//关闭监听线程
            thread.close();
            thread = null;
        }
        
        //3.主动关闭回调
        service.onClose(this, CLOSE_SELF);
    }

    /****************************************************************************************/
    //获取参数
    /****************************************************************************************/

    /** 支付业务侧调用发送PING */
    public boolean sendPing()
    {
        return send(new WebSocketPacket().setMaskingKey(maskingKey).setPing());
    }
    
    /** 支持业务侧发送断开数据包 */
    public boolean sendClose(int code)
    {
        return send(new WebSocketPacket().setMaskingKey(maskingKey).setClose(code));
    }

    /** 支持业务侧发送文本数据包 */
    public boolean sendText(String text)
    {
        return send(new WebSocketPacket().setMaskingKey(maskingKey).setText(text));
    }
    
    /** 支持业务侧发送二进制数据包 */
    public boolean sendBinary(byte[] binary)
    {
        return send(new WebSocketPacket().setMaskingKey(maskingKey).setBinary(binary));
    }

    /** 发送到流中 */
    private boolean send(WebSocketPacket packet)
    {
        try
        {
            byte[] data = packet.toPacket();
            synchronized (this)
            {
                output.write(data, 0, data.length);
            }
            
            return true;
        }
        catch (IOException e)
        {//异常要求先关闭监听线程再通知业务侧
            thread.close();
            service.onExceptionSend(this, e, packet);
            return false;
        }
    }
    
    /****************************************************************************************/
    //获取结果
    /****************************************************************************************/
    
    /** 获取响应状态 */
    public int getResponseStatus()
    {
        return responseStatus;
    }
    
    /** 判断响应是否成功 */
    public boolean isResponseSuccess()
    {
        return responseStatus == 101;
    }
    
    /** 获取响应内容(正确内容或错误内容) */
    public String getResponseText()
    {
        return responseText;
    }
    
    /** 获取响应状态行 */
    public String getResponseStatusLine()
    {//HTTP/1.1 101 Switching+Protocols
        return responseStatusLine;
    }
    
    public String getProtocol()
    {
        return requestPropertyMap.get(_SEC_WEBSOCKET_PROTOCOL_);
    }
    
    /** 获取响应属性 */
    public String getResponseProperty(String key)
    {
        return responsePropertyMap.get(key.toLowerCase());
    }
    
    /** 是否有响应属性 */
    public boolean hasResponseProperty(String key)
    {
        return responsePropertyMap.containsKey(key.toLowerCase());
    }
    
    /** 获取响应属性表 */
    public HashMapSS getResponsePropertyMap()
    {
        return responsePropertyMap;
    }
    
    /** 获取响应属性列表 */
    public List<String> getResponseList()
    {
        return new ArrayList<String>(responsePropertyMap.values());
    }

    public WebSocketService getService()
    {
        return service;
    }
    
    /************************************************************************/
    //监听HTTP升级为WS后的消息
    /************************************************************************/

    private class WebSocketThread extends Threadx
    {
        private WebSocket websocket;
        
        public WebSocketThread(WebSocket websocket)
        {
            this.websocket = websocket;
        }
        
        @Override
        protected String getThreadName()
        {
            return "WebSocket";
        }
        
        @Override
        protected void loop()
        {
            try
            {
                WebSocketPacket packet = WebSockets.parse(input);
                switch (packet.opcode)
                {
                case PONG:
                {//心跳响应置心跳检测数为0
                    handshakeNum.getAndSet(0);
                    service.onPong(websocket);
                    break;
                }
                case PING:
                {//心跳请求回复响应
                    send(new WebSocketPacket().setMaskingKey(maskingKey).setPong());
                    service.onPing(websocket);
                    break;
                }
                case CLOSE:
                {//接收到断开，关闭连接进入重连机制
                    close();
                    service.onClose(websocket, packet.getCloseCode());
                    break;
                }
                case TEXT:
                {//接收到文本消息
                    service.onText(websocket, packet.getText());
                    break;
                }
                case BIN:
                {//接收到二进制流
                    service.onBinary(websocket, packet.getData());
                    break;
                }
                case MID:
                {//接收到中间包
                    service.onMiddle(websocket, packet.getData());
                    break;
                }
                default:
                {//最后为其他自定义的报文
                    service.onOther(websocket, packet.getData());
                    break;
                }
                }
            }
            catch (SocketTimeoutException e) 
            {//超时继续监听
            }
            catch (Throwable e)
            {//异常要求先关闭监听线程再通知业务侧
                close();
                service.onException(websocket, e);
            }
        }
        
        @Override
        protected void closeAfter()
        {
            responseStatus = 0;
            responseText = null;
            Closes.closeIgnoreException(input, output, socket);
        }
    }
    
    /****************************************************************************************/
    //定时检查任务
    /****************************************************************************************/
    
    private class WebSocketTask implements Task
    {
        private WebSocket websocket;
        
        public WebSocketTask(WebSocket websocket)
        {
            this.websocket = websocket;
        }
        
        @Override
        public void execute()
        {
            if (!websocket.thread.isRunning())
            {//未启动的尝试启动
                websocket.execute();
            }
            else if (websocket.handshakeNum.get() >= websocket.handshakeMaxNum)
            {//已启动但握手失败则关闭，等待下次重连
                websocket.thread.close();
                websocket.service.onClose(websocket, CLOSE_HANDSHAKE);
            }
            else
            {//已启动正常的发送握手消息
                websocket.sendPing();
            }
        }
    }
}
